/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.query.predicate;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.PriorityQueue;
import java.util.logging.Logger;

public abstract class Ranges {
  /**
   * An empty range (no numbers on the number line)
   */
  public static class EmptyRange extends Ranges.Range {
    public final static Ranges.EmptyRange INSTANCE = new EmptyRange();

    private EmptyRange() {}

    @Override
    public Ranges.Range intersect(final Ranges.Range other) {
      return INSTANCE;
    }

    @Override
    public EmptyRange mirror() {
      return INSTANCE;
    }

    @Override
    public String toString() {
      return "[EMPTY]";
    }
  }

  /**
   * A compact representation for a subset of the natural integers.
   */
  public abstract static class Range {
    public static Ranges.Range intersect(final Ranges.Range one,
                                         final Ranges.Range two) {
      return one.intersect(two);
    }

    public static SimpleRange over(final int start, final int end) {
      return new SimpleRange(start, end);
    }

    public static Range union(final SimpleRange... over) {
      return RangeUnion.makeUnion(Arrays.asList(over));
    }

    public abstract Ranges.Range intersect(Ranges.Range other);

    public abstract Range mirror();
  }

  /**
   * A union of disjoint simple ranges of integers.
   */
  public static class RangeUnion extends Ranges.Range {
    private static Ranges.Range makeUnion(final List<Ranges.SimpleRange> ranges) {
      PriorityQueue<Ranges.SimpleRange> rangeQ =
          new PriorityQueue<Ranges.SimpleRange>();
      List<Ranges.SimpleRange> nonIntersecting = new ArrayList<Ranges.SimpleRange>();
      for (Ranges.SimpleRange range : ranges) {
        rangeQ.add(range);
      }
      Ranges.SimpleRange range = rangeQ.poll();
      while (rangeQ.size() > 0) {
        Ranges.SimpleRange range2 = rangeQ.poll();
        if (range.overlaps(range2)) {
          range = range.union(range2);
        } else {
          nonIntersecting.add(range);
          range = range2;
        }
      }
      nonIntersecting.add(range);
      if (nonIntersecting.size() == 1) {
        return nonIntersecting.get(0);
      } else {
        return new RangeUnion(nonIntersecting);
      }
    }

    private final List<Ranges.SimpleRange> simpleRanges;

    public RangeUnion(final List<Ranges.SimpleRange> ranges) {
      this.simpleRanges = ranges;
    }

    private Ranges.Range intersect(final List<Ranges.SimpleRange> otherRanges) {
      List<Ranges.SimpleRange> intersections = new ArrayList<Ranges.SimpleRange>();
      for (Ranges.SimpleRange rangeOuter : otherRanges) {
        for (Ranges.SimpleRange rangeInner : this.simpleRanges) {
          Ranges.Range intersection = rangeOuter.intersect(rangeInner);
          if (intersection != EmptyRange.INSTANCE) {
            intersections.add((SimpleRange) intersection);
          }
        }
      }
      if (intersections.size() == 0) {
        return EmptyRange.INSTANCE;
      } else if (intersections.size() == 1) {
        return intersections.get(0);
      } else {
        return RangeUnion.makeUnion(intersections);
      }
    }

    @Override
    public Ranges.Range intersect(final Ranges.Range other) {
      if (other == EmptyRange.INSTANCE) { return other; }
      if (other instanceof Ranges.RangeUnion) { return this
          .intersect(((Ranges.RangeUnion) other).simpleRanges); }
      if (other instanceof Ranges.SimpleRange) {
        List<Ranges.SimpleRange> listOfOther = new ArrayList<Ranges.SimpleRange>();
        listOfOther.add((Ranges.SimpleRange) other);
        return this.intersect(listOfOther);
      }
      String msg = "Internal Error: Unepxected case.";
      PositionPredicates.LOG.severe(msg);
      throw new RuntimeException(msg);
    }

    @Override
    public RangeUnion mirror() {
      List<SimpleRange> newUnion = new ArrayList<SimpleRange>();
      for (SimpleRange range : this.simpleRanges) {
        newUnion.add(range.mirror());
      }
      return new RangeUnion(newUnion);
    }

    @Override
    public String toString() {
      String res = "U{";
      for (SimpleRange r : this.simpleRanges) {
        res += r.toString() + " ";
      }
      return res.trim() + "}";
    }
  }

  /**
   * A simple range of integers on a number line.
   */
  public static class SimpleRange extends Ranges.Range implements
      Comparable<Ranges.SimpleRange> {
    private final int end;
    private final int start;

    protected SimpleRange(final int start, final int end) {
      this.start = start;
      this.end = end;
    }

    @Override
    public int compareTo(final Ranges.SimpleRange other) {
      int comp = Integer.compare(this.start, other.start);
      if (comp == 0) {
        comp = Integer.compare(this.end, other.end);
      }
      return comp;
    }

    @Override
    public Ranges.Range intersect(final Ranges.Range other) {
      if (other == EmptyRange.INSTANCE) { return other; }
      if (other instanceof Ranges.RangeUnion) { return other.intersect(this); }
      if (other instanceof Ranges.SimpleRange) {
        int newStart = Math.max(this.start, ((Ranges.SimpleRange) other).start);
        int newEnd = Math.min(this.end, ((Ranges.SimpleRange) other).end);
        if (newStart <= newEnd) {
          return new SimpleRange(newStart, newEnd);
        } else {
          return EmptyRange.INSTANCE;
        }
      }
      String msg = "Internal Error: Unepxected case.";
      LOG.severe(msg);
      throw new RuntimeException(msg);
    }

    @Override
    public SimpleRange mirror() {
      int newStart =
          this.end == Integer.MAX_VALUE ? Integer.MIN_VALUE
              : this.end == Integer.MIN_VALUE ? Integer.MAX_VALUE : -this.end;
      int newEnd =
          this.start == Integer.MAX_VALUE ? Integer.MIN_VALUE
              : this.start == Integer.MIN_VALUE ? Integer.MAX_VALUE : -this.start;
      return new SimpleRange(newStart, newEnd);
    }

    public boolean overlaps(final Ranges.SimpleRange other) {
      // .....|-----------this---------|........ //
      // ..|--case1--|.|--case2--|.|--case3--|.. //
      // ..|--------------case4--------------|.. //
      // other contains the start of this (case1 and case4)
      if (this.start >= other.start && this.start <= other.end) { return true; }
      // this contains the start of other (case2 and case3)
      if (other.start >= this.start && other.start <= this.end) { return true; }
      return false;
    }

    @Override
    public String toString() {
      return String.format("[%s,%s]", printNum(this.start), printNum(this.end));
    }

    public Ranges.SimpleRange union(final Ranges.SimpleRange other) {
      assert this.overlaps(other);
      int newStart = Math.min(this.start, other.start);
      int newEnd = Math.max(this.end, other.end);
      return new SimpleRange(newStart, newEnd);
    }
  }

  private final static Logger LOG = Logger.getLogger(Ranges.class.getName());

  public static String printNum(final int num) {
    if (num == Integer.MAX_VALUE) {
      return "+inf";
    } else if (num == Integer.MIN_VALUE) { return "-inf"; }
    return "" + num;
  }
}